const fs = require('fs')
const EventsEmitter = require('events')
const Queue = require('./linked-queue')

class MyWriteStream extends EventsEmitter {
  constructor(path, options = {}) {
    super()
    this.path = path
    this.flags = options.flags || 'w'
    this.mode = options.mode || 438
    this.autoClose = options.autoClose || true
    this.start = options.start || 0
    this.encoding = options.encoding || 'utf-8'
    this.highWaterMark = options.highWaterMark || 16 * 1024

    this.open()

    this.writeOffset = this.start
    this.writing = false
    this.writeLen = 0 // 累计写入量
    this.needDrain = false
    this.cache = new Queue()
  }

  open() {
    // 原生 fs.open
    fs.open(this.path, this.flags, (err, fd) => {
      if (err) {
        this.emit('error')
      }
      // 正常打开文件
      this.fd = fd
      this.emit('open', fd)
    })
  }

  write(chunk, encoding, cb) {
    chunk = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)

    this.writeLen += chunk.length

    // 写入量小于最大容量
    let flag = this.writeLen < this.highWaterMark
    // 否则需要 drain
    this.needDrain = !flag

    if (this.writing) {
      // 当前正在执行写入, 内容正在排队
      this.cache.enQueue({ chunk, encoding, cb })
    } else {
      this.writing = true
      // 当前不是正在写入那么就执行写入
      //  回调: 切片的思想 (除了执行传入的callback, 还要执行 清空内存 的操作)
      this._write(chunk, encoding, () => {
        cb()
        // 清空排队的内容
        this._clearBuffer()
      })
    }
    return flag
  }

  // 真正的写入操作
  _write(chunk, encoding, cb) {
    // 同步的open,完成后还没回归主线程
    if (typeof this.fd != 'number') {
      // 使用发布订阅确保文件已经被打开
      return this.once('open', () => {
        return this._write(chunk, encoding, cb)
      })
    }
    fs.write(this.fd, chunk, this.start, chunk.length, this.writeOffset, (err, written) => {
      this.writeOffset += written
      this.writeLen -= written

      cb && cb()
    })
  }
  // 回调的切片: 清空内存
  _clearBuffer() {
    let data = this.cache.deQueue()
    if (data) {
      this._write(data.element.chunk, data.element.encoding, () => {
        data.element.cb && data.element.cb()
        this._clearBuffer()
      })
    } else {
      if (this.needDrain) {
        this.needDrain = false
        this.emit('drain')
      }
    }
  }
}

const ws = new MyWriteStream('test.txt', { highWaterMark: 1 })
ws.on('open', (fd) => {
  console.info('open------>', fd)
})

ws.on('error', () => {
  console.info('error------>')
})

for (let i = 0; i < 3; i++) {
  ws.write(`Content-${i}`, 'utf-8', () => {
    console.info(`Content-${i}  OK`)
  })
}

ws.on('drain', () => {
  console.info('drain 事件被触发了')
})
